package com.job.admin.job.jobbean;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.JobKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.quartz.QuartzJobBean;

import com.job.admin.common.CustmerJobConst;
import com.job.admin.job.enmus.RegistryConfig;
import com.job.admin.job.model.JobInfoModel;
import com.job.admin.job.schedule.JobDynamicScheduler;
import com.job.admin.job.thread.JobFailMonitorHelper;
import com.job.admin.job.thread.JobRegistryMonitorHelper;
import com.job.admin.jobGroup.model.JobGroupModel;
import com.job.admin.log.model.JobLogModel;
import com.job.admin.menu.util.DateUtil;
import com.job.core.biz.ExecutorBiz;
import com.job.core.biz.model.ReturnModel;
import com.job.core.biz.model.TriggerParam;
import com.job.core.enums.ExecutorBlockStrategyEnum;
import com.job.core.enums.ExecutorFailStrategyEnum;
import com.job.core.route.ExecutorRouteStrategyEnum;
import com.job.core.rpc.NetComClientProxy;
/**
 * 
 * @Title: RemoteHttpJobBean.java
 * @Package: com.job.admin.job.jobbean
 * @Description: 加载远程job
 * @author: sunwei
 * @date: 2017年8月6日 下午9:55:04
 * @version: V1.0
 */
public class RemoteHttpJobBean extends QuartzJobBean {
    private static Logger logger = LoggerFactory.getLogger(RemoteHttpJobBean.class);

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
	JobKey jobKey = context.getTrigger().getJobKey();
	Long jobId = Long.valueOf(jobKey.getName());
	JobInfoModel jobInfo = JobDynamicScheduler.jobDao.queryJobById(jobId);
	// log part-1
	JobLogModel jobLog = new JobLogModel();
	jobLog.setJobGroup(jobInfo.getJobGroup());
	jobLog.setJobId(jobInfo.getId());
	JobDynamicScheduler.logDao.save(jobLog);
	logger.info(">>>>>>>>>>> job trigger start, jobId:{}",jobLog.getId());
	// log part-2 param
	// jobLog.setExecutorAddress(executorAddress);
	jobLog.setGlueType(jobInfo.getGlueType());
	jobLog.setExecutorHandler(jobInfo.getExecutorHandler());
	jobLog.setExecutorParam(jobInfo.getExecutorParam());
	jobLog.setTriggerTime(new Date());
	// trigger request
	TriggerParam triggerParam = new TriggerParam();
	triggerParam.setJobId(jobInfo.getId());
	triggerParam.setExecutorHandler(jobInfo.getExecutorHandler());
	triggerParam.setExecutorParams(jobInfo.getExecutorParam());
	triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy());
	triggerParam.setGlueType(jobInfo.getGlueType());
	triggerParam.setGlueSource(jobInfo.getGlueSource());
	triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdateTime().getTime());
	triggerParam.setLogId(jobLog.getId());
	triggerParam.setLogDateTim(jobLog.getTriggerTime().getTime());
	// do trigger
	ReturnModel<String> triggerResult = doTrigger(triggerParam, jobInfo, jobLog);

	// fail retry
	if (triggerResult.getCode() == ReturnModel.FAILUE_CODE && ExecutorFailStrategyEnum.match(
			jobInfo.getExecutorFailStrategy(), null) == ExecutorFailStrategyEnum.FAIL_RETRY) {
	    ReturnModel<String> retryTriggerResult = doTrigger(triggerParam,
		    jobInfo, jobLog);

	    triggerResult.setCode(retryTriggerResult.getCode());
	    triggerResult
		    .setMsg(triggerResult.getMsg()
			    + "<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>失败重试<<<<<<<<<<< </span><br><br>"
			    + retryTriggerResult.getMsg());
	}

	// log part-2
	jobLog.setTriggerCode(Long.valueOf(triggerResult.getCode()));
	jobLog.setTriggerMsg(triggerResult.getMsg());
	JobDynamicScheduler.logDao.updateTriggerInfo(jobLog);

	// monitor triger
	JobFailMonitorHelper.monitor(Integer.valueOf(jobLog.getId().toString()));
	logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}",jobLog.getId());
    }
    
    private ReturnModel<String> doTrigger(TriggerParam triggerParam, JobInfoModel jobInfo, JobLogModel jobLog){
	StringBuffer triggerSb = new StringBuffer();
	 Map<String, String> returnMap=new HashMap<String, String>();
	// exerutor address list
	ArrayList<String> addressList = null;
	JobGroupModel group =JobDynamicScheduler.jobGroupDao.queryJobGoupById(jobInfo.getJobGroup());
	if (group.getAddressType() == 0) {
		triggerSb.append("注册方式：自动注册");
		addressList = (ArrayList<String>) JobRegistryMonitorHelper.discover(RegistryConfig.RegistType.EXECUTOR.name(), group.getAppName());
	} else {
		triggerSb.append("注册方式：手动录入");
		if (StringUtils.isNotBlank(group.getAddressList())) {
			addressList = new ArrayList<String>(Arrays.asList(group.getAddressList().split(",")));
		}
	}
	triggerSb.append("<br>阻塞处理策略：").append(ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION).getTitle());
	triggerSb.append("<br>失败处理策略：").append(ExecutorFailStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorFailStrategyEnum.FAIL_ALARM).getTitle());
	triggerSb.append("<br>地址列表：").append(addressList!=null?addressList.toString():"");
	if (CollectionUtils.isEmpty(addressList)) {
		triggerSb.append("<br>----------------------<br>").append("调度失败：").append("执行器地址为空");
		return new ReturnModel<String>(ReturnModel.FAILUE_CODE, triggerSb.toString());
	}
	// trigger remote executor
	if (addressList.size() == 1) {
		String address = addressList.get(0);
		jobLog.setExecutorAddress(address);

		ReturnModel<String> runResult = runExecutor(triggerParam, address);
		triggerSb.append("<br>----------------------<br>").append(runResult.getMsg());

		return new ReturnModel<String>(runResult.getCode(), triggerSb.toString());
	} else {
		// executor route strategy
		ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null);
		triggerSb.append("<br>路由策略：").append(executorRouteStrategyEnum!=null?(executorRouteStrategyEnum.name() + "-" + executorRouteStrategyEnum.getTitle()):null);
		if (executorRouteStrategyEnum == null) {
			triggerSb.append("<br>----------------------<br>").append("调度失败：").append("执行器路由策略为空");
			return new ReturnModel<String>(ReturnModel.FAILUE_CODE, triggerSb.toString());
		}
		if (executorRouteStrategyEnum != ExecutorRouteStrategyEnum.FAILOVER) {
			// get address
			String address = executorRouteStrategyEnum.getRouter().route(jobInfo.getId(), addressList);
			jobLog.setExecutorAddress(address);

			// run
			ReturnModel<String> runResult = runExecutor(triggerParam, address);
			triggerSb.append("<br>----------------------<br>").append(runResult.getMsg());

			return new ReturnModel<String>(runResult.getCode(), triggerSb.toString());
		} else {
			for (String address : addressList) {
				// beat
				ReturnModel<String> beatResult = beatExecutor(address);
				triggerSb.append("<br>----------------------<br>").append(beatResult.getMsg());

				if (beatResult.getCode() == ReturnModel.SUCCESS_CODE) {
					jobLog.setExecutorAddress(address);

					ReturnModel<String> runResult = runExecutor(triggerParam, address);
					triggerSb.append("<br>----------------------<br>").append(runResult.getMsg());

					return new ReturnModel<String>(runResult.getCode(), triggerSb.toString());
				}
			}
			return new ReturnModel<String>(ReturnModel.FAILUE_CODE, triggerSb.toString());
		}
	}
}
    public ReturnModel<String> runExecutor(TriggerParam triggerParam, String address){
	ReturnModel<String> runResult = null;
	try {
		ExecutorBiz executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address).getObject();
		runResult = executorBiz.run(triggerParam);
	} catch (Exception e) {
		logger.error("", e);
		runResult = new ReturnModel<String>(ReturnModel.FAILUE_CODE, ""+e );
	}

	StringBuffer sb = new StringBuffer("触发调度：");
	sb.append("<br>address：").append(address);
	sb.append("<br>code：").append(runResult.getCode());
	sb.append("<br>msg：").append(runResult.getMsg());
	runResult.setMsg(sb.toString());

	return runResult;
    }
    
    public ReturnModel<String> beatExecutor(String address){
	ReturnModel<String> beatResult = null;
	try {
		ExecutorBiz executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address).getObject();
		beatResult = executorBiz.beat();
	} catch (Exception e) {
		logger.error("", e);
		beatResult = new ReturnModel<String>(ReturnModel.FAILUE_CODE, ""+e );
	}

	StringBuffer sb = new StringBuffer("心跳检测：");
	sb.append("<br>address：").append(address);
	sb.append("<br>code：").append(beatResult.getCode());
	sb.append("<br>msg：").append(beatResult.getMsg());
	beatResult.setMsg(sb.toString());

	return beatResult;
    }

}
